#include "StreamManager.h"
#include "RealMediaObj.h"
#include "VodMediaObj.h"
#include "VoiceIcomObj.h"
#include "FilterFactory.h"
#include "FileReaderFactory.h"

static const char *http_end = "\r\n\r\n";

CStreamManager::CStreamManager()
{
	signal(SIGPIPE, SIG_IGN);
}

CStreamManager::~CStreamManager()
{

}

int CStreamManager::StartService(int nPort)
{
	//注册消息回调函数
	CXMessageQueue::Instance()->RegistMessage(MSG_MEDIA, CStreamManager::OnMessage, this);

	return Start(nPort);
}

int CStreamManager::StopService()
{
	//取消消息回调函数
	CXMessageQueue::Instance()->ReleaseMessage(MSG_MEDIA);

	return Stop();
}

int CStreamManager::OnAccept(int nSocket)
{
	m_clientMap[nSocket] = NULL;

	return J_OK;
}

int CStreamManager::OnRead(int nSocket)
{
	std::map<int, J_MediaObj *>::iterator it = m_clientMap.find(nSocket);
	if (it == m_clientMap.end())
	{
		J_OS::LOGINFO("CStreamManager::OnRead No Client");
		return J_NOT_EXIST;
	}

	if (ParserRequest(nSocket, it->second) < 0)
	{
		return J_UNKNOW;
	}

	return J_OK;
}

int CStreamManager::OnWrite(int nSocket)
{
	std::map<int, J_MediaObj *>::iterator it = m_clientMap.find(nSocket);
	if (it == m_clientMap.end())
	{
		return J_NOT_EXIST;
	}

	int nRet = J_OK;
	J_MediaObj *pClient = dynamic_cast<J_MediaObj *>(it->second);
	if (pClient != NULL)
	{
		nRet = pClient->Process(J_IoWrite);
		if (nRet < 0)
			J_OS::LOGERROR("CStreamManager::OnWrite Error");
	}
	else
	{
		usleep(1);
		//J_OS::LOGINFO("CStreamManager::OnWrite No Client");
		//return J_NOT_EXIST;
		return J_OK;
	}

	return nRet;
}

int CStreamManager::OnBroken(int nSocket)
{
	std::map<int, J_MediaObj *>::iterator it = m_clientMap.find(nSocket);
	if (it == m_clientMap.end())
	{
		return J_NOT_EXIST;
	}

	if (it->second != NULL)
	{
		it->second->Clearn();
		delete it->second;
		it->second = NULL;
	}

	m_clientMap.erase(it);

	return J_OK;
}

int CStreamManager::GetSocketByResid(const char *pResid)
{
	ResidMap::iterator it = m_residMap.find(pResid);
	if (it != m_residMap.end())
	{
		if (!it->second.empty())
		{
			int nSocket = it->second.back();
			it->second.pop_back();

			if (it->second.empty())
				m_residMap.erase(it);

			return nSocket;
		}
		return 0;
	}

	return 0;
}

int CStreamManager::ParserRequest(int nSocket, J_MediaObj *pClient)
{
	int nRet = J_OK;
	J_OS::CTCPSocket readSocket(nSocket);
	char read_buff[1024] = {0};
	int read_ret = 0;
	int totle_recv = 0;
	do
	{
		read_ret = readSocket.Read(read_buff + totle_recv, sizeof(read_buff));
		if (read_ret < 0)
		{
			return J_SOCKET_ERROR;
		}
		if (strstr(read_buff, "HTTP") == NULL && strstr(read_buff, "RTSP") == NULL
            && strstr(read_buff, "JOSP") == NULL)
			return J_OK;

		totle_recv += read_ret;
	} while(strstr(read_buff, http_end) == NULL);

	//printf(read_buff);
	J_RequestFilter *protocolFilter = CFilterFactory::Instance()->GetFilter(nSocket, read_buff);
	if (protocolFilter == NULL)
	{
		return J_PARAM_ERROR;
	}

	protocolFilter->Parser(read_buff);
	J_FileReader *fileReader = CFileReaderFactory::Instance()->GetFileReader(nSocket, read_buff);
	//printf("status = %d\n", protocolFilter->GetStatus());
	if (fileReader != NULL)
	{
		if (!(protocolFilter->GetStatus() & J_ProParamOk))
		{
			J_VodCommandFilter *vodTimeCommand = dynamic_cast<J_VodCommandFilter *>((J_Obj *)protocolFilter);
			fileReader->SetTime(vodTimeCommand->GetBeginTime(), vodTimeCommand->GetEndTime());

			J_MediaContext *pMediaContext = NULL;
			fileReader->GetContext(pMediaContext);
			protocolFilter->SetContext(pMediaContext);
			if (pMediaContext)
				free(pMediaContext);
			protocolFilter->SetContext(NULL);
		}
		/*else if (protocolFilter->GetStatus() & DATA_UNRADY)
		{
			J_VodCommandFilter *vodTimeCommand = dynamic_cast<J_VodCommandFilter *>((J_Obj *)protocolFilter);
			fileReader->SetTime(vodTimeCommand->GetBeginTime());
		}*/
	}

	if (!(protocolFilter->GetStatus() & J_ProConnOk))
	{
		char retBuff[1024] = {0};
		int retLen = 0;
		protocolFilter->GetRetrunVal(retBuff, retLen);
		if (readSocket.Write_n((char *)retBuff, retLen) < 0)
		{
			if (protocolFilter)
				CFilterFactory::Instance()->DelFilter(nSocket);
			if (fileReader)
				CFileReaderFactory::Instance()->DelFileReader(nSocket);

			J_OS::LOGERROR("CStreamManager::ParserRequest Send Header error");
			return J_SOCKET_ERROR;
		}

		return J_OK;
	}
	else
	{
		nRet = ProcessCommand(nSocket, protocolFilter, fileReader, pClient);
	}

	return nRet;
}

int CStreamManager::ProcessCommand(int nSocket, J_Obj *pObj, J_FileReader *pReader, J_MediaObj *pClient)
{
	int nRet = J_OK;
	J_CommandFilter *videoCommand = dynamic_cast<J_CommandFilter *>(pObj);
	if (videoCommand != NULL)
	{
		switch (videoCommand->GetCommandType())
		{
		case J_START_REAL:
		case J_START_VOD:
		case J_START_VOICE:
			if (pClient == NULL)
			{
				switch (videoCommand->GetCommandType())
				{
				case J_START_REAL:
					{
						m_clientMap[nSocket] = new CRealMediaObj(nSocket, videoCommand->GetStreamType(), pObj);
						std::string resid = videoCommand->GetResid();
						ResidMap::iterator it = m_residMap.find(resid);
						if (it == m_residMap.end())
						{
							std::vector<int> vecResid;
							m_residMap[resid] = vecResid;
							m_residMap[resid].push_back(nSocket);
						}
						else
						{
							it->second.push_back(nSocket);
						}
					}
					break;
				case J_START_VOD:
					m_clientMap[nSocket] = new CVodMediaObj(nSocket, pObj, pReader);
					break;
				case J_START_VOICE:
					m_clientMap[nSocket] = new CVoiceIcomObj(nSocket, pObj);
					break;
				default:
					assert(false);
					break;
				}
				pClient = m_clientMap[nSocket];
			}
		}
	}

	nRet = pClient->Process(J_IoRead);
	if (nRet < 0)
	{
		J_OS::LOGINFO("CStreamManager::ProcessCommand Process Error");
		return nRet;
	}

	char retBuff[1024] = {0};
	int retLen = 0;

	J_RequestFilter *protocolFilter = dynamic_cast<J_RequestFilter *>(pObj);
	protocolFilter->GetRetrunVal(retBuff, retLen);

	//J_OS::LOGINFO(retBuff);
	J_OS::CTCPSocket writeSocket(nSocket);
	if (writeSocket.Write_n((char *)retBuff, retLen) < 0)
	{
		J_OS::LOGERROR("CStreamManager::ProcessCommand Send Header error");
		return J_SOCKET_ERROR;
	}

	pClient->Run();
	if (IS_CLOSE_CMD(videoCommand->GetCommandType()))
		return J_ON_CLOSE;

	return J_OK;
}

void CStreamManager::ProcMessage(BaseMessage *pMessage)
{
	MessageImpl<std::string> *pMsgImpl = dynamic_cast<MessageImpl<std::string> *>(pMessage);
	if (pMsgImpl != NULL)
	{
		int nSocket = 0;
		CMediaConvert mediaConvert;
		while ((nSocket = GetSocketByResid(
				pMsgImpl->OnMessage(mediaConvert).c_str())) > 0)
		{
			RECLock(m_locker);
			Broken(nSocket, m_evListen);
			VecSocket::iterator it = m_vecSocket.begin();
			for (; it != m_vecSocket.end(); it++)
			{
				if ((*it).fd == nSocket)
				{
					m_vecSocket.erase(it);
					break;
				}
			}
			RECUnlock(m_locker);
		}
	}
	delete pMessage;
}

